#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <sys/time.h>
#include <signal.h>
#include "msg_queue_peer.h"
#include "shmem.h"
#include "cJSON.h"
#include "MQTTClient.h"

#define CLIENTID "ExampleClientPub" 
#define TOPIC "up"  // 发布主题
#define TOPICS "lc" // 订阅主题
#define QOS 1
#define TIMEOUT 10000L
#define MSG_PATH "/tmp/ipc/msgqueue/peer/" // 消息队列路径
#define MAGIC_ID 'j'                       // keyID
char mes[1024];
char ADDR[64];
char *p = NULL;

MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient_deliveryToken token;
int rc;
int ch;
union val_t
{
    BOOL b_val;  // bool类型存储空间
    int i_val;   // 整形值存储空间
    float f_val; // 浮点值存储空间
};

// 消息队列 控制下位机结构体
struct control
{
    long Equ;        // 消息类型
    int key;         // 设备ID
    char name[32];   // 设备名称
    union val_t val; // 数据设置
    int addr;        // 线圈寄存器
    int type;        // 数据点值的类型，1：bool类型 2：int型  3：float型
};

// 32数据结构体
struct data32
{
    int key;
    char name[32];
    union val_t val; // 实时数据
    int type;        // 数据类型
};
struct data32 d32;

// modbus数据结构体
struct datam
{
    int key;
    char name[32];
    union val_t val; // 实时数据
    int addr;        // modbus地址
    int type;        // 数据类型
};
struct datam dm;

// 共享内存存放数据结构体
struct data
{
    struct data32 d32[10];
    struct datam dmod[10];
    int arrsize;
};
struct data da;

// 上报方式结构体
struct report
{
    int type;   // 0-不上报，客户端主动采集；1-变化上报，即连续2次值不相等；2-周期上报
    int period; // 上报周期时间，单位秒，仅在type=2时有效
};
struct report re;
struct shm_param shm;  //共享内存结构体
struct data *saddr;    //接收共享内存地址
int updata = 0;        //刷新上报标志位
char *payloadptr; // 接受到的消息内容
cJSON *root;      //反序列化根节点
volatile MQTTClient_deliveryToken deliveredtoken;

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

// 修改点表上报模式
void changeMode(const char *file_path, int new_type, int new_period)
{
    // 打开文件读取内容
    FILE *file = fopen(file_path, "r");
    if (!file)
    {
        perror("Failed to open file");
        return;
    }

    // 获取文件大小
    fseek(file, 0, SEEK_END);
    long file_size = ftell(file);
    rewind(file);

    // 分配内存读取文件内容
    char *file_content = (char *)malloc(file_size + 1);
    if (!file_content)
    {
        perror("Failed to allocate memory");
        fclose(file);
        return;
    }
    fread(file_content, 1, file_size, file);
    file_content[file_size] = '\0'; // 确保字符串以 '\0' 结尾
    fclose(file);

    // 解析 JSON 数据
    cJSON *root = cJSON_Parse(file_content);
    if (!root)
    {
        printf("Failed to parse JSON: %s\n", cJSON_GetErrorPtr());
        free(file_content);
        return;
    }

    // 修改 "type" 和 "period" 值
    cJSON *report = cJSON_GetObjectItem(root, "report");
    if (report)
    {
        cJSON *type = cJSON_GetObjectItem(report, "type");
        cJSON *period = cJSON_GetObjectItem(report, "period");

        if (type)
        {
            cJSON_SetNumberValue(type, new_type);
        }
        else
        {
            printf("Field 'type' not found in 'report'\n");
        }

        if (period)
        {
            cJSON_SetNumberValue(period, new_period);
        }
        else
        {
            printf("Field 'period' not found in 'report'\n");
        }
    }
    else
    {
        printf("Field 'report' not found\n");
    }

    // 序列化修改后的 JSON 数据
    char *modified_content = cJSON_Print(root);
    if (!modified_content)
    {
        printf("Failed to serialize modified JSON\n");
        cJSON_Delete(root);
        free(file_content);
        return;
    }

    // 写回文件
    file = fopen(file_path, "w");
    if (!file)
    {
        perror("Failed to open file for writing");
        cJSON_Delete(root);
        free(file_content);
        free(modified_content);
        return;
    }
    fwrite(modified_content, 1, strlen(modified_content), file);
    fclose(file);

    // 释放内存
    cJSON_Delete(root);
    free(file_content);
    free(modified_content);

    printf("Successfully modified 'type' to %d and 'period' to %d\n", new_type, new_period);
}

// 序列化共享内存数据
char *sendmsg()
{
    char str[32];
    // 创建根对象
    cJSON *root = cJSON_CreateObject();
    // 添加字段 "type"
    cJSON_AddNumberToObject(root, "type", 1);
    // 添加字段 "result"
    cJSON_AddNumberToObject(root, "result", 0);
    // 创建数组 "data"
    cJSON *data_array = cJSON_CreateArray();
    // 添加第一个数据点
    cJSON *data_item1 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item1, "key", saddr->d32[0].key);
    snprintf(str, sizeof(str), "%.2f", saddr->d32[0].val.f_val);
    cJSON_AddStringToObject(data_item1, "val", str);
    cJSON_AddItemToArray(data_array, data_item1);
    // 添加第二个数据点
    cJSON *data_item2 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item2, "key", saddr->d32[1].key);
    snprintf(str, sizeof(str), "%.2f", saddr->d32[1].val.f_val);
    cJSON_AddStringToObject(data_item2, "val", str);
    cJSON_AddItemToArray(data_array, data_item2);
    // 添加第三个数据点
    cJSON *data_item3 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item3, "key", saddr->d32[2].key);
    snprintf(str, sizeof(str), "%.2f", saddr->d32[2].val.f_val);
    cJSON_AddStringToObject(data_item3, "val", str);
    cJSON_AddItemToArray(data_array, data_item3);
    // 添加第四个数据点
    cJSON *data_item4 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item4, "key", saddr->d32[3].key);
    snprintf(str, sizeof(str), "%.2f", saddr->d32[3].val.f_val);
    cJSON_AddStringToObject(data_item4, "val", str);
    cJSON_AddItemToArray(data_array, data_item4);
    // 添加第五个数据点
    cJSON *data_item5 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item5, "key", saddr->d32[4].key);
    snprintf(str, sizeof(str), "%d", saddr->d32[4].val.b_val);
    cJSON_AddStringToObject(data_item5, "val", str);
    cJSON_AddItemToArray(data_array, data_item5);
    // 添加第六个数据点
    cJSON *data_item6 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item6, "key", saddr->d32[5].key);
    snprintf(str, sizeof(str), "%d", saddr->d32[5].val.b_val);
    cJSON_AddStringToObject(data_item6, "val", str);
    cJSON_AddItemToArray(data_array, data_item6);
    // 添加第七个数据点
    cJSON *data_item7 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item7, "key", saddr->d32[6].key);
    snprintf(str, sizeof(str), "%d", saddr->d32[6].val.b_val);
    cJSON_AddStringToObject(data_item7, "val", str);
    cJSON_AddItemToArray(data_array, data_item7);
    // 添加第八个数据点
    cJSON *data_item8 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item8, "key", saddr->d32[7].key);
    snprintf(str, sizeof(str), "%d", saddr->d32[7].val.b_val);
    cJSON_AddStringToObject(data_item8, "val", str);
    cJSON_AddItemToArray(data_array, data_item8);
    // 添加第九个数据点
    cJSON *data_item9 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item9, "key", saddr->d32[7].key);
    snprintf(str, sizeof(str), "%d", saddr->d32[7].val.b_val);
    cJSON_AddStringToObject(data_item9, "val", str);
    cJSON_AddItemToArray(data_array, data_item9);

    // 添加第十个数据点
    cJSON *data_item10 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item10, "key", saddr->dmod[0].key);
    snprintf(str, sizeof(str), "%.2f", saddr->dmod[0].val.f_val);
    cJSON_AddStringToObject(data_item10, "val", str);
    cJSON_AddItemToArray(data_array, data_item10);
    // 添加第11个数据点
    cJSON *data_item11 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item11, "key", saddr->dmod[1].key);
    snprintf(str, sizeof(str), "%.2f", saddr->dmod[1].val.f_val);
    cJSON_AddStringToObject(data_item11, "val", str);
    cJSON_AddItemToArray(data_array, data_item11);

    // 添加第12个数据点
    cJSON *data_item12 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item12, "key", saddr->dmod[2].key);
    snprintf(str, sizeof(str), "%d", saddr->dmod[2].val.b_val);
    cJSON_AddStringToObject(data_item12, "val", str);
    cJSON_AddItemToArray(data_array, data_item12);

    // 添加第13个数据点
    cJSON *data_item13 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item13, "key", saddr->dmod[3].key);
    snprintf(str, sizeof(str), "%.2f", saddr->dmod[3].val.f_val);
    cJSON_AddStringToObject(data_item13, "val", str);
    cJSON_AddItemToArray(data_array, data_item13);

    // 添加第14个数据点
    cJSON *data_item14 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item14, "key", saddr->dmod[4].key);
    snprintf(str, sizeof(str), "%d", saddr->dmod[4].val.b_val);
    cJSON_AddStringToObject(data_item14, "val", str);
    cJSON_AddItemToArray(data_array, data_item14);

    // 添加第15个数据点
    cJSON *data_item15 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item15, "key", saddr->dmod[5].key);
    snprintf(str, sizeof(str), "%d", saddr->dmod[5].val.b_val);
    cJSON_AddStringToObject(data_item15, "val", str);
    cJSON_AddItemToArray(data_array, data_item15);

    // 添加第16个数据点
    cJSON *data_item16 = cJSON_CreateObject();
    cJSON_AddNumberToObject(data_item16, "key", saddr->dmod[6].key);
    snprintf(str, sizeof(str), "%d", saddr->dmod[6].val.b_val);
    cJSON_AddStringToObject(data_item16, "val", str);
    cJSON_AddItemToArray(data_array, data_item16);

    // 将数组添加到根对象
    cJSON_AddItemToObject(root, "data", data_array);
    // 序列化为字符串
    char *msg = cJSON_Print(root);
    cJSON_Delete(root);
    return msg;
}

// 定时器函数
void timer_handler(int signum)
{
    char *msg = sendmsg();
    printf("Serialized JSON:\n%s\n", msg);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    pubmsg.payload = msg;
    pubmsg.payloadlen = (int)strlen(msg);
    // MQTT发布消息
    MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);
    // 释放内存
    free(msg);
    alarm(re.period);
}
// 重启进程函数
void update()
{
    char buf[32] = {};
    char *exec_argv[] = {"./test", buf, 0};
    int count = 0;
    printf("Runing:%s %d\n", "./test", count);
    snprintf(buf, sizeof(buf), "%d", count + 1);
    execv("/proc/self/exe", exec_argv);
}
// MQTT数据订阅
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;

    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: ");

    payloadptr = message->payload;

    root = cJSON_Parse(payloadptr);
    cJSON *type = cJSON_GetObjectItem(root, "type");
    updata = type->valueint;
    // 控制指令
    if (type->valueint == 2)
    {
        cJSON *data = cJSON_GetObjectItem(root, "data");
        cJSON *key = cJSON_GetObjectItem(data, "key");
        cJSON *val = cJSON_GetObjectItem(data, "val");
        if (key->valueint <= 309 && key->valueint >= 301)
        {
            struct control mod;
            mod.Equ = 32;
            mod.key = key->valueint;
            if (key->valueint <= 304 && key->valueint >= 301)
            {
                mod.val.f_val = atof(val->valuestring);
            }
            else
            {
                mod.val.b_val = atoi(val->valuestring);
            }
            msg_queue_send("msg", &mod, sizeof(mod), 0);
        }
        else if (key->valueint <= 107 && key->valueint >= 101)
        {

            struct control stm;
            stm.Equ = 1;
            stm.key = key->valueint;
            if ((key->valueint <= 102 && key->valueint >= 101) || key->valueint == 104)
            {
                stm.val.f_val = atof(val->valuestring);
            }
            else
            {
                stm.val.b_val = atoi(val->valuestring);
            }
            msg_queue_send("msg", &stm, sizeof(stm), 0);
        }
    }
    // 模式转换
    else if (type->valueint == 3)
    {
        alarm(0);
        cJSON *data = cJSON_GetObjectItem(root, "data");
        cJSON *type = cJSON_GetObjectItem(data, "type");
        cJSON *period = cJSON_GetObjectItem(data, "period");
        changeMode("./node.json", type->valueint, period->valueint);
        update();
    }
    for (i = 0; i < message->payloadlen; i++)
    {
        putchar(*payloadptr++);
    }
    putchar('\n');

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}
// 点表反序列化并放置在共享内存
void pointTable()
{
    saddr = (struct data *)shm_getaddr(&shm);
    FILE *fp = NULL;
    // 检查文件是否存在且可读
    while (access("./node.json", F_OK) == -1 || access("./node.json", R_OK) == -1)
    {
        printf("Waiting for node.json file to become available...\n");
        sleep(1); // 等待 1 秒钟，避免过度占用 CPU
    }

    // 现在文件可以读取了，打开文件
    fp = fopen("./node.json", "r");
    if (fp == NULL)
    {
        perror("Failed to open node.json");
        return;
    }
    char buf[10248];
    fread(buf, sizeof(buf), 1, fp);

    // 反序列化上报类型
    cJSON *root = cJSON_Parse(buf);
    cJSON *item;
    cJSON *report = cJSON_GetObjectItem(root, "report");
    cJSON *type = cJSON_GetObjectItem(report, "type");
    cJSON *period = cJSON_GetObjectItem(report, "period");
    re.type = type->valueint;
    re.period = period->valueint;

    cJSON *mqtt_server = cJSON_GetObjectItem(root, "mqtt_server");
    cJSON *addr = cJSON_GetObjectItem(mqtt_server, "addr");
    cJSON *port = cJSON_GetObjectItem(mqtt_server, "port");

    snprintf(ADDR, sizeof(ADDR), "%s:%d", addr->valuestring, port->valueint);
    // 32结构体反序列化
    cJSON *stm32 = cJSON_GetObjectItem(root, "stm32");
    cJSON *data = cJSON_GetObjectItem(stm32, "data");
    da.arrsize = cJSON_GetArraySize(data);
    for (int i = 0; i < da.arrsize; i++)
    {
        item = cJSON_GetArrayItem(data, i);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        cJSON *name = cJSON_GetObjectItem(item, "name");
        type = cJSON_GetObjectItem(item, "type");
        da.d32[i].key = key->valueint;
        strcpy(da.d32[i].name, name->valuestring);
        da.d32[i].type = type->valueint;
        saddr->d32[i].key = key->valueint;
        strcpy(saddr->d32[i].name, name->valuestring);
        saddr->d32[i].type = type->valueint;
    }

    // modbus结构体反序列化
    cJSON *modbus = cJSON_GetObjectItem(root, "modbus");
    data = cJSON_GetObjectItem(modbus, "data");
    da.arrsize = cJSON_GetArraySize(data);
    for (int i = 0; i < da.arrsize; i++)
    {
        item = cJSON_GetArrayItem(data, i);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        cJSON *name = cJSON_GetObjectItem(item, "name");
        addr = cJSON_GetObjectItem(item, "addr");
        type = cJSON_GetObjectItem(item, "type");
        da.dmod[i].key = key->valueint;
        saddr->dmod[i].key = key->valueint;
        strcpy(da.dmod[i].name, name->valuestring);
        strcpy(saddr->dmod[i].name, name->valuestring);
        da.dmod[i].addr = addr->valueint;
        saddr->dmod[i].addr = addr->valueint;
        da.dmod[i].type = type->valueint;
        saddr->dmod[i].type = type->valueint;
    }
}

int main(int argc, char *argv[])
{
    shm_init(&shm, "shm", sizeof(da));
    saddr = (struct data *)shm_getaddr(&shm);
    // 点表反序列化并放置在共享内存
    pointTable();
    // 周期上报
    if (re.type == 2)
    {
        alarm(re.period);
        signal(SIGALRM, timer_handler);
    }
    // MQTT客户端创建
    MQTTClient_create(&client, ADDR, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    // 回调函数
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }

    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n",
           TOPICS, CLIENTID, QOS);

    MQTTClient_subscribe(client, TOPICS, QOS);

    // 序列化发送
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    // 判断两次是否相同
    char *msghistory = sendmsg();
    char *msg;
    do
    {
        // 刷新上报
        while (re.type == 0)
        {
            // 刷新上报数据
            if (updata == 1)
            {
                cJSON *limit = cJSON_GetObjectItem(root, "limit");
                if (!strcmp(limit->valuestring, "all"))
                {
                    char *msg = sendmsg();
                    printf("Serialized JSON:\n%s\n", msg);
                    pubmsg.payload = msg;
                    pubmsg.payloadlen = (int)strlen(msg);
                    // MQTT发布消息
                    MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);
                    // 释放内存
                    free(msg);
                    updata = 0;
                }
            }
        }
        //变化上报
        while (re.type == 1)
        {
            sleep(1);
            msg = sendmsg();
            if (strcmp(msg, msghistory))
            {
                pubmsg.payload = msg;
                pubmsg.payloadlen = (int)strlen(msg);
                // MQTT发布消息
                MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);
                msghistory = msg;
            }
        }

    } while (1);
    printf("Waiting for up to %d seconds for publication of %s\n"
           "on topic %s for client with ClientID: %s\n",
           (int)(TIMEOUT / 1000), p, TOPIC, CLIENTID);
    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);

    printf("Message with delivery token %d delivered\n", token);

    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    free(p);
    return rc;
}